package wap;

import java.util.concurrent.Callable;

import wap.dco.Log;
import wap.dco.factory.ILogFactory;
import wap.dco.factory.impl.LogFactoryImpl;
import wap.io.BufferedFileWriter;
import wap.io.BufferedGzipReader;
import wap.queue.IQueue;

public class ThreadConsum extends ThreadAbstract implements Callable<String> {

	private IQueue<String> queue;

	public IQueue<String> getQueue() {
		return queue;
	}

	public void setQueue(IQueue<String> queue) {
		this.queue = queue;
	}

	@Override
	public String call() throws Exception {

		int tryCount = 0;

		while (true) {

			if (tryCount >= 10) {
				System.out.println(Thread.currentThread().getName() + "-->is break!!!");
				break;
			}

			int num = 0;
			String fileName = this.queue.get();
			ILogFactory factory = new LogFactoryImpl();
			if (fileName == null) {
				tryCount++;
				Thread.sleep(3 * 1000);
			} else {
				try {

					tryCount = 0;
					BufferedGzipReader br = new BufferedGzipReader(fileName);
					BufferedFileWriter bw = new BufferedFileWriter("D:/test/" + br.getName());
					String line = null;
					System.out.println("Get-->" + Thread.currentThread().getName() + "-->" + fileName
					        + "-->QueueSize=" + this.queue.queueSize() + "-->HashSize="
					        + this.queue.hashSize());
					while ((line = br.readLine()) != null) {
						Log log = factory.createLog(line);
						if (log == null)
							continue;
						num++;
						StringBuffer sb = new StringBuffer();

						sb.append(TimeFormat.format(log.getTime()));
						sb.append("\t");
						sb.append(log.getDomain());
						sb.append("\t");
						sb.append(log.getUrl());
						sb.append("\t");
						sb.append(log.getType());
						sb.append("\n");
						bw.write(sb.toString());
					}
					bw.flush();
					bw.close();
					br.close();
					System.out.println("Deal-->" + Thread.currentThread().getName() + fileName + "-->num="
					        + num);
					// Thread.sleep(10 * 1000);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		return null;
	}
}
